package com.yuzhi.master.deviceData;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSONObject;
import com.ctiot.aep.mqmsgpush.sdk.IMsgConsumer;
import com.ctiot.aep.mqmsgpush.sdk.IMsgListener;
import com.ctiot.aep.mqmsgpush.sdk.MqMsgConsumer;
import com.yuzhi.common.contant.Constants;
import com.yuzhi.common.core.domain.entity.SysDictData;
import com.yuzhi.master.drainRRmst.domain.vo.DrainRRmstVo;
import com.yuzhi.master.drainRRmst.service.IDrainRRmstService;
import com.yuzhi.master.drainStPptnr.service.DrainStPptnrService;
import com.yuzhi.master.pump.mapper.DrainRRmstMapper;
import com.yuzhi.master.pump.service.IDrainPumpBjService;
import com.yuzhi.master.sewage.device.service.IDrainDeviceZtService;
import com.yuzhi.master.sewage.rmstSz.service.DrainRmstSzService;
import com.yuzhi.system.domain.request.SysDictDataReq;
import com.yuzhi.system.service.ISysDictDataService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.SmartLifecycle;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executor;

/**
 * @Author ARUNNE
 * @Date 2025-08-05 15:44
 */
@Slf4j
@Component
public class ReceiveDeviceDataHandler implements SmartLifecycle {

    private volatile boolean running = false;
    private IMsgConsumer consumer;
    private final Executor messageProcessingExecutor;

    @Autowired
    private IDrainRRmstService drainRRmstService;

    @Autowired
    private DrainRmstSzService drainRmstSzService;

    @Autowired
    private DrainStPptnrService drainStPptnrService;

    @Autowired
    private IDrainDeviceZtService drainDeviceZtService;

    @Autowired
    private ISysDictDataService sysDictDataService;

    @Autowired
    private IDrainPumpBjService iDrainPumpBjService;

    private final String server = "pulsar://192.168.14.43:17650";
    private final String tenantId = "300";
    private final String token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIzMDBfMSJ9.603H0lwAlM8lhRjUbM68Rg-0YLlfFR-fzukHgbwt3bw";
    private final String certFilePath = "";
    //设备上下线设备标识
    private final String DEVICE_STATUS = "deviceOnlineOfflineReport";

    public ReceiveDeviceDataHandler(DrainStPptnrService drainStPptnrService, DrainRRmstMapper drainRRmstMapper, IDrainRRmstService drainRRmstService) {

        // 创建专门处理消息的线程池
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("mqtt-msg-processor-");
        executor.initialize();
        this.messageProcessingExecutor = executor;
    }

    private void initializeAndStartConsumer() {

        // 创建消息监听器（使用线程池异步处理）
        IMsgListener msgListener = msg -> {
                    if (!running) {
                        return;
                    }
                    messageProcessingExecutor.execute(() -> {
                        if (!running) {
                            return;
                        }
                    try {
                        // 记录接收到的设备消息到日志
                        log.info("接收到设备消息：{}", msg);
                        // 1. 解析消息为JSON对象并提取设备ID
                        JSONObject json = JSONObject.parseObject(msg);
                        String deviceIds = json.getString("deviceId");
                        //存储日志
                        //drainRRmstService.insertFacilitylog(deviceIds, msg);
                        //判断deviceId设备id是否带有下划线 对提取的deviceId 设备id进行解析
                        String deviceId = parseDeviceId(deviceIds);
                        // 2. 查询设备类型字典数据
                        // 构建查询条件：设备ID、字典类型为设备类型、未删除状态
                        SysDictDataReq sysDictDataReq = new SysDictDataReq();
                        sysDictDataReq.setDictLabel(deviceId);
                        sysDictDataReq.setDictType("equipment_device_type");
                        sysDictDataReq.setDeleted(Constants.UNDELETED.byteValue());
                        // 执行查询字典
                        List<SysDictData> sysDictDataList = sysDictDataService.selectDictDataList(sysDictDataReq);

                        // 3. 存在字典数据时进行业务处理
                        if (!sysDictDataList.isEmpty()) {
                            log.info("查询字典类型成功！");
                            SysDictData sysDictData = sysDictDataList.get(0);

                            if (ObjectUtil.isNotEmpty(sysDictData)) {
                                // 3.1 获取设备类型（字典键值）
                                String deviceType = sysDictData.getDictValue();

                                // 3.2 查询设备相关信息
                                DrainRRmstVo drainRRmstVo = drainRRmstService.queryById(deviceId);

                                // 3.3 记录设备日志
                                //drainRRmstService.insertFacilitylog(deviceId, msg);

                                // 3.4 处理设备状态消息（上线/下线）
                                String messageType = json.getString("messageType");
                                if (DEVICE_STATUS.equals(messageType)) {
                                    drainDeviceZtService.saveDeviceZtByMqtt(msg, drainRRmstVo);
                                }
                                // 3.5 根据设备类型进行对应业务处理
                                // 雨量设备（类型4）
                                if ("4".equals(deviceType)) {
                                    drainStPptnrService.saveMessage(msg);
                                }
                                // 3.6 流量设备（类型2）
                                else if ("2".equals(deviceType)) {
                                    drainRmstSzService.install(msg, drainRRmstVo);
                                }
                                // 3.7 水质设备（类型1）
                                else if ("1".equals(deviceType)) {
                                    drainRmstSzService.saveSz(msg, drainRRmstVo);
                                }
                                // 3.8 泵站设备（类型5）
                                /*else if ("5".equals(deviceType)) {
                                    iDrainPumpBjService.saveBjMessage(deviceId, msg);
                                }*/
                            }
                        }

                    } catch (Exception e) {
                        // 记录处理异常日志
                        log.error("处理设备消息时发生异常，消息内容：{}，异常信息：{}", msg, e.getMessage(), e);
                        e.printStackTrace();
                    }
                });
                };

        consumer = new MqMsgConsumer();
        try {
            // saas_datacollect为订阅主题
            consumer.init(server, tenantId, token, certFilePath, Arrays.asList("saas_datacollect"), msgListener);
            consumer.start();
            System.out.println("MQTT consumer started successfully.");

        } catch (Exception e) {
            e.printStackTrace();
            // 失败后尝试重启
            if (running) {
                restartConsumer();
            }
        }
    }

    private void restartConsumer() {
        try {
            if (!running) {
                return;
            }
            if (consumer != null) {
                consumer.stop();
                consumer.destroy();
            }
            if (running) {
                initializeAndStartConsumer();
            }
        } catch (Exception e) {
            e.printStackTrace();
            // 重试逻辑
            try {
                Thread.sleep(5000);
                if (running) {
                    restartConsumer();
                }
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            }
        }
    }

    @Override
    public void start() {
        System.out.println("开始监听设备消息...");
        if (!this.running) {
            this.running = true;
            initializeAndStartConsumer();
        }
    }

    @Override
    public void stop() {
        if (this.running) {
            this.running = false;
            if (consumer != null) {
                try {
                    consumer.stop();
                    consumer.destroy();
                    System.out.println("结束监听设备消息...");
                    // 关闭消息处理线程池，防止关闭数据源后仍有任务访问数据库
                    try {
                        if (messageProcessingExecutor instanceof ThreadPoolTaskExecutor) {
                            ((ThreadPoolTaskExecutor) messageProcessingExecutor).shutdown();
                        }
                    } catch (Exception ex) {
                        log.warn("关闭消息处理线程池异常", ex);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public boolean isRunning() {
        return this.running;
    }

    @Override
    public int getPhase() {
        return Integer.MAX_VALUE;
    }

    //判断deviceId设备id是否带有下划线 对提取的deviceId 设备id进行解析
    public static String parseDeviceId(String deviceId) {
        if (deviceId == null || !deviceId.contains("_")) {
            return deviceId;
        }
        // 获取下划线前的部分
        String prefix = deviceId.substring(0, deviceId.indexOf('_'));

        // 找到最后一个数字的位置
        int lastDigitIndex = -1;
        for (int i = 0; i < prefix.length(); i++) {
            if (Character.isDigit(prefix.charAt(i))) {
                lastDigitIndex = i;
            } else {
                // 遇到非数字字符时停止查找
                break;
            }
        }

        // 提取数字后的内容
        return (lastDigitIndex + 1 < prefix.length())
                ? prefix.substring(lastDigitIndex + 1)
                : "";  // 如果没有非数字内容则返回空字符串
    }
}
